package com.bw.yk10.app;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.bw.yk10.func.MyDeserializationSchema;
import com.bw.yk10.util.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Properties;

public class Text01 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

//        2、原始数据层ODS：新建Maven工程，添加Flink相关依赖，按照数仓分层创建表，编写Flink流式程序，使用Flink CDC技术，实时捕获MySQL表线上问诊业务数据，并过滤脏数据和解析json，实时存储Kafka消息队列：ods-medical-topic

//        Properties properties = new Properties();
//        properties.setProperty("decimal.handling.mode","double");

        DebeziumSourceFunction<String> mysqlSource=MySQLSource.<String>builder()
        .hostname("192.168.18.105")
         .port(3306)
         .databaseList("medical")
         .username("root")
         .password("root")
         .deserializer(new MyDeserializationSchema())
//         .debeziumProperties(properties)
         .startupOptions(StartupOptions.initial()).build();

        DataStreamSource<String> stream=env.addSource(mysqlSource);

        SingleOutputStreamOperator<String> filterDS=stream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {

                if(s==null||s.length()<1||s.equals(""))
                    return false;

                try {
                    JSONObject jsonObject=JSONObject.parseObject(s);
                    if(jsonObject!=null&&jsonObject.get("data")!=null){
                        return true;
                    }
                }catch (Exception e)
                {
                    e.printStackTrace();
                }
                return false;
            }
        });

        filterDS.print();
//        filterDS.addSink(new MyKafkaUtil().getKafkaProducer("ods-medical-topic"));



        env.execute();

    }
}
